{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {
    "colab_type": "text",
    "id": "-thKr9rm_2rB"
   },
   "source": [
    "# Install Dependencies\n",
    "\n",
    "If you are running on Google Colab, you need to install the necessary dependencies before beginning the exercise."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/",
     "height": 479
    },
    "colab_type": "code",
    "id": "zCb7eB-l_-HH",
    "outputId": "0a563fe6-5981-4482-fad7-7f7b46e2a379"
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "NOTE: Intentionally crashing session to use the newly installed library.\n",
      "\n",
      "\u001b[33mWARNING: Skipping pyarrow as it is not installed.\u001b[0m\n",
      "Requirement already satisfied: ray[debug]==0.7.5 in /usr/local/lib/python3.6/dist-packages (0.7.5)\n",
      "Requirement already satisfied: jsonschema in /usr/local/lib/python3.6/dist-packages (from ray[debug]==0.7.5) (2.6.0)\n",
      "Requirement already satisfied: numpy>=1.14 in /usr/local/lib/python3.6/dist-packages (from ray[debug]==0.7.5) (1.16.5)\n",
      "Requirement already satisfied: pytest in /usr/local/lib/python3.6/dist-packages (from ray[debug]==0.7.5) (3.6.4)\n",
      "Requirement already satisfied: redis>=3.3.2 in /usr/local/lib/python3.6/dist-packages (from ray[debug]==0.7.5) (3.3.10)\n",
      "Requirement already satisfied: funcsigs in /usr/local/lib/python3.6/dist-packages (from ray[debug]==0.7.5) (1.0.2)\n",
      "Requirement already satisfied: six>=1.0.0 in /usr/local/lib/python3.6/dist-packages (from ray[debug]==0.7.5) (1.12.0)\n",
      "Requirement already satisfied: colorama in /usr/local/lib/python3.6/dist-packages (from ray[debug]==0.7.5) (0.4.1)\n",
      "Requirement already satisfied: protobuf>=3.8.0 in /usr/local/lib/python3.6/dist-packages (from ray[debug]==0.7.5) (3.10.0)\n",
      "Requirement already satisfied: click in /usr/local/lib/python3.6/dist-packages (from ray[debug]==0.7.5) (7.0)\n",
      "Requirement already satisfied: filelock in /usr/local/lib/python3.6/dist-packages (from ray[debug]==0.7.5) (3.0.12)\n",
      "Requirement already satisfied: pyyaml in /usr/local/lib/python3.6/dist-packages (from ray[debug]==0.7.5) (3.13)\n",
      "Requirement already satisfied: py-spy; extra == \"debug\" in /usr/local/lib/python3.6/dist-packages (from ray[debug]==0.7.5) (0.2.2)\n",
      "Requirement already satisfied: psutil; extra == \"debug\" in /usr/local/lib/python3.6/dist-packages (from ray[debug]==0.7.5) (5.4.8)\n",
      "Requirement already satisfied: setproctitle; extra == \"debug\" in /usr/local/lib/python3.6/dist-packages (from ray[debug]==0.7.5) (1.1.10)\n",
      "Requirement already satisfied: atomicwrites>=1.0 in /usr/local/lib/python3.6/dist-packages (from pytest->ray[debug]==0.7.5) (1.3.0)\n",
      "Requirement already satisfied: setuptools in /usr/local/lib/python3.6/dist-packages (from pytest->ray[debug]==0.7.5) (41.2.0)\n",
      "Requirement already satisfied: py>=1.5.0 in /usr/local/lib/python3.6/dist-packages (from pytest->ray[debug]==0.7.5) (1.8.0)\n",
      "Requirement already satisfied: more-itertools>=4.0.0 in /usr/local/lib/python3.6/dist-packages (from pytest->ray[debug]==0.7.5) (7.2.0)\n",
      "Requirement already satisfied: pluggy<0.8,>=0.5 in /usr/local/lib/python3.6/dist-packages (from pytest->ray[debug]==0.7.5) (0.7.1)\n",
      "Requirement already satisfied: attrs>=17.4.0 in /usr/local/lib/python3.6/dist-packages (from pytest->ray[debug]==0.7.5) (19.2.0)\n",
      "Requirement already satisfied: bs4 in /usr/local/lib/python3.6/dist-packages (0.0.1)\n",
      "Requirement already satisfied: beautifulsoup4 in /usr/local/lib/python3.6/dist-packages (from bs4) (4.6.3)\n"
     ]
    }
   ],
   "source": [
    "print('NOTE: Intentionally crashing session to use the newly installed library.\\n')\n",
    "\n",
    "!pip uninstall -y pyarrow\n",
    "!pip install ray[debug]==0.7.5\n",
    "!pip install bs4\n",
    "\n",
    "# A hack to force the runtime to restart, needed to include the above dependencies.\n",
    "import os\n",
    "os._exit(0)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "colab_type": "text",
    "id": "l6wpgwlHdfus"
   },
   "source": [
    "# Exercise 1 - Simple Parallelization\n",
    "\n",
    "**GOAL:** The goal of this exercise is to show how to run simple tasks in parallel.\n",
    "\n",
    "This script is running too slowly, but the computation is embarrassingly parallel. In this exercise, you will use Ray to execute the functions in parallel to speed it up.\n",
    "\n",
    "### Introduction to Remote Functions\n",
    "\n",
    "The `@ray.remote` decorator turns a Python function into a \"remote function\" that Ray can run in parallel. Here is a simple example:\n",
    "\n",
    "```python\n",
    "# A regular Python function.\n",
    "def regular_function():\n",
    "    return 1\n",
    "\n",
    "# A Ray remote function.\n",
    "@ray.remote\n",
    "def remote_function():\n",
    "    return 1\n",
    "```\n",
    "\n",
    "There are a few key differences between the original function and the decorated one:\n",
    "\n",
    "1. **Invocation:** The regular version is called with `regular_function()`, whereas the remote version is called with `remote_function.remote()`.\n",
    "2. **Return values:** `regular_function` executes synchronously and returns the result of the function (`1`), whereas `remote_function` immediately returns an `ObjectID` (a future) and then executes the task in the background on a separate worker process. The result of the future can be obtained by calling `ray.get` on the `ObjectID`.\n",
    "    ```python\n",
    "    >>> regular_function()\n",
    "    1\n",
    "    \n",
    "    >>> remote_function.remote()\n",
    "    ObjectID(1c80d6937802cd7786ad25e50caf2f023c95e350)\n",
    "    \n",
    "    >>> ray.get(remote_function.remote())\n",
    "    1\n",
    "    ```\n",
    "3. **Parallelism:** Invocations of `regular_function` happen **serially**:\n",
    "    ```python\n",
    "    # These are executed one at a time, back-to-back.\n",
    "    result = 0\n",
    "    for _ in range(4):\n",
    "        result += regular_function()\n",
    "    assert result == 4\n",
    "    ```\n",
    "    In contrast, invocations of `remote_function` happen in **parallel**:\n",
    "    ```python\n",
    "    # Executing these functions happens at the same time in the background, and we get the results using ray.get.\n",
    "    results = []\n",
    "    for _ in range(4):\n",
    "        results.append(remote_function.remote())\n",
    "    assert sum(ray.get(results)) == 4\n",
    "    ```"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/",
     "height": 34
    },
    "colab_type": "code",
    "id": "xLJx-XpJdfuu",
    "outputId": "2699d8f9-ace8-4662-a869-1c390b46c39f"
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Successfully imported ray!\n"
     ]
    }
   ],
   "source": [
    "from __future__ import absolute_import\n",
    "from __future__ import division\n",
    "from __future__ import print_function\n",
    "\n",
    "import ray\n",
    "import time\n",
    "\n",
    "print('Successfully imported ray!')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "colab_type": "text",
    "id": "PvwwpldHdfux"
   },
   "source": [
    "Start the processes that make up the Ray runtime. By default, Ray does not schedule more tasks concurrently than there are CPUs, but this example requires four tasks to run concurrently, so we tell Ray that there are four CPUs. In practice, you would usually just let Ray detect the number of CPUs on the machine.\n",
    "\n",
    "`ignore_reinit_error=True` just suppresses errors if the cell is run multiple times."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/",
     "height": 34
    },
    "colab_type": "code",
    "id": "1MjLO5L4dfux",
    "outputId": "46d3b188-11ee-492c-e1dc-dc2534aaf2ec"
   },
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "2019-10-13 10:06:37,698\tERROR worker.py:1432 -- Calling ray.init() again after it has already been called.\n"
     ]
    }
   ],
   "source": [
    "ray.init(num_cpus=4, ignore_reinit_error=True)\n",
    "\n",
    "# Sleep a little to improve the accuracy of the timing measurements used below,\n",
    "# because some workers may still be starting up in the background.\n",
    "time.sleep(2.0)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "colab_type": "text",
    "id": "D35JdJQ8dfu1"
   },
   "source": [
    "**EXERCISE:** The loop below takes too long, but the four function calls could be executed in parallel. This should speed up the execution from four seconds to one second. \n",
    "\n",
    "First, turn `slow_function` into a remote function, then execute the tasks in parallel by calling `slow_function.remote()` and obtaining the results with `ray.get` on the list of returned object IDs."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/",
     "height": 68
    },
    "colab_type": "code",
    "id": "pnNgl3cddfu2",
    "outputId": "df898887-dd1a-41ca-88d6-46447c82c6c2"
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Executing the for loop took 1.013 seconds.\n",
      "The results are: [0, 1, 2, 3]\n",
      "Run the next cell to check if the exercise was performed correctly.\n"
     ]
    }
   ],
   "source": [
    "# A function simulating a more interesting computation that takes one second.\n",
    "@ray.remote\n",
    "def slow_function(i):\n",
    "    time.sleep(1)\n",
    "    return i\n",
    "\n",
    "start_time = time.time()\n",
    "\n",
    "results = []\n",
    "for i in range(4):\n",
    "  results.append(slow_function.remote(i))\n",
    "  \n",
    "results = ray.get(results)\n",
    "  \n",
    "duration = time.time() - start_time\n",
    "print('Executing the for loop took {:.3f} seconds.'.format(duration))\n",
    "print('The results are:', results)\n",
    "print('Run the next cell to check if the exercise was performed correctly.')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "colab_type": "text",
    "id": "r_FfKxocdfu3"
   },
   "source": [
    "**VERIFY:** Run some checks to verify that the changes you made to the code were correct. Some of the checks should fail when you initially run the cells. After completing the exercises, the checks should pass."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/",
     "height": 34
    },
    "colab_type": "code",
    "id": "IlrIrAyldfu4",
    "outputId": "65f5c3c2-c161-471e-e2ee-45e333bfb2cb"
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Success! The example took 1.0131359100341797 seconds.\n"
     ]
    }
   ],
   "source": [
    "assert results == [0, 1, 2, 3], 'Did you remember to call ray.get?'\n",
    "assert duration < 1.1, ('The loop took {:.3f} seconds. This is too slow.'\n",
    "                        .format(duration))\n",
    "assert duration > 1, ('The loop took {:.3f} seconds. This is too fast.'\n",
    "                      .format(duration))\n",
    "\n",
    "print('Success! The example took {} seconds.'.format(duration))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "colab_type": "text",
    "id": "AC1rwahidfu5"
   },
   "source": [
    "**EXERCISE:** Use the UI to view the task timeline and to verify that the four tasks were executed in parallel. You can do this as follows.\n",
    "\n",
    "1. Run the following cell to generate a JSON file containing the profiling data.\n",
    "2. Download the timeline file by right clicking on `exercise_1.json` in the **Files** tab in the navigator to the left, right clicking, and selecting  **\"Download\"**.\n",
    "3. Enter **chrome://tracing** into the Chrome web browser, click on the **\"Load\"** button, and select the downloaded JSON file.\n",
    "\n",
    "To navigate within the timeline:\n",
    "- Move around by clicking and dragging.\n",
    "- Zoom in and out by holding **alt** on Windows or **option** on Mac and scrolling.\n",
    "\n",
    "**NOTE:** The timeline visualization will only work in **Chrome**."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "colab": {},
    "colab_type": "code",
    "id": "LmYOA4sGdfu6"
   },
   "outputs": [],
   "source": [
    "ray.timeline(filename=\"exercise_1.json\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "colab_type": "text",
    "id": "TD9T8jPYd2bO"
   },
   "source": [
    "# Exercise 2 - Parallel Data Processing with Task Dependencies\n",
    "\n",
    "**GOAL:** The goal of this exercise is to pass object IDs into remote functions to encode dependencies between tasks.\n",
    "\n",
    "In this exercise, we construct a sequence of tasks, each of which depends on the previous, mimicking a data parallel application. Within each sequence, tasks are executed serially, but multiple sequences of tasks can be executed in parallel. You will use Ray to speed up the computation by parallelizing the sequences.\n",
    "\n",
    "### Concept for this Exercise - Task Dependencies\n",
    "\n",
    "Consider the following basic remote function that returns the argument passed to it. If we pass in some normal Python objects, the results returned by `ray.get` should be the same objects.\n",
    "\n",
    "```python\n",
    "@ray.remote\n",
    "def f(x):\n",
    "    return x\n",
    "\n",
    ">>> x1_id = f.remote(1)\n",
    ">>> ray.get(x1_id)\n",
    "1\n",
    "\n",
    ">>> x2_id = f.remote([1, 2, 3])\n",
    ">>> ray.get(x2_id)\n",
    "[1, 2, 3]\n",
    "```\n",
    "\n",
    "However, **object IDs** can also be passed into remote functions. When the function is executed, Ray will automatically substitute the underlying Python object that the object ID refers to. In a sense, it's the same as calling `ray.get` on each argument that's passed in as an argument.\n",
    "\n",
    "```python\n",
    ">>> y1_id = f.remote(x1_id)\n",
    ">>> ray.get(y1_id)\n",
    "1\n",
    "\n",
    ">>> y2_id = f.remote(x2_id)\n",
    ">>> ray.get(y2_id)\n",
    "[1, 2, 3]\n",
    "```\n",
    "\n",
    "When implementing a remote function, the function should expect a regular Python object regardless of whether the caller passes in a regular Python object or an object ID.\n",
    "\n",
    "**These task dependencies affect scheduling.** In the example above, the task that creates `y1_id` depends on the task that creates `x1_id`. This means that:\n",
    "\n",
    "- The second task will not be executed until the first task has finished executing.\n",
    "- If the two tasks are scheduled on different machines, the output of the first task (the value corresponding to `x1_id`) will be copied over the network to the machine where the second task is scheduled."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "colab": {},
    "colab_type": "code",
    "id": "jRayjahDd-AU"
   },
   "outputs": [],
   "source": [
    "from __future__ import absolute_import\n",
    "from __future__ import division\n",
    "from __future__ import print_function\n",
    "\n",
    "import numpy as np\n",
    "import time\n",
    "\n",
    "import ray"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/",
     "height": 34
    },
    "colab_type": "code",
    "id": "csUGAU13d-Zs",
    "outputId": "d0e7ffa8-e884-485c-9525-65f1510802b7"
   },
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "2019-10-13 10:06:52,014\tERROR worker.py:1432 -- Calling ray.init() again after it has already been called.\n"
     ]
    }
   ],
   "source": [
    "ray.init(num_cpus=4, ignore_reinit_error=True)\n",
    "\n",
    "# Sleep a little to improve the accuracy of the timing measurements used below,\n",
    "# because some workers may still be starting up in the background.\n",
    "time.sleep(2.0)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "colab_type": "text",
    "id": "dGrF31cceDMk"
   },
   "source": [
    "**EXERCISE:** Below are some helper functions that mimic the pattern of a data parallel application that we want to speed up. To do so, you'll need to turn all of these functions into remote functions. Remember that you don't need to worry about whether the caller passes in an object ID or a regular object, because in both cases the arguments will be regular objects when the function executes. This means that even if you pass in an object ID, you **do not need to call `ray.get`** inside of these remote functions."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "colab": {},
    "colab_type": "code",
    "id": "ZU1nRG8beACP"
   },
   "outputs": [],
   "source": [
    "@ray.remote\n",
    "def load_data(filename):\n",
    "    time.sleep(0.1)\n",
    "    return np.ones((1000, 100))\n",
    "\n",
    "@ray.remote\n",
    "def normalize_data(data):\n",
    "    time.sleep(0.1)\n",
    "    return data - np.mean(data, axis=0)\n",
    "\n",
    "@ray.remote\n",
    "def extract_features(normalized_data):\n",
    "    time.sleep(0.1)\n",
    "    return np.hstack([normalized_data, normalized_data ** 2])\n",
    "\n",
    "@ray.remote\n",
    "def compute_loss(features):\n",
    "    num_data, dim = features.shape\n",
    "    time.sleep(0.1)\n",
    "    return np.sum((np.dot(features, np.ones(dim)) - np.ones(num_data)) ** 2)\n",
    "\n",
    "assert hasattr(load_data, 'remote'), 'load_data must be a remote function'\n",
    "assert hasattr(normalize_data, 'remote'), 'normalize_data must be a remote function'\n",
    "assert hasattr(extract_features, 'remote'), 'extract_features must be a remote function'\n",
    "assert hasattr(compute_loss, 'remote'), 'compute_loss must be a remote function'"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "colab_type": "text",
    "id": "TlF-_kc2eHc4"
   },
   "source": [
    "**EXERCISE:** The loop below takes too long. Parallelize the four passes through the loop by turning `load_data`, `normalize_data`, `extract_features`, and `compute_loss` into remote functions and then retrieving the losses with `ray.get`.\n",
    "\n",
    "**NOTE:** You should only use **ONE** call to `ray.get`. For example, the object ID returned by `load_data` should be passed directly into `normalize_data` without needing to be retrieved by the driver."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/",
     "height": 68
    },
    "colab_type": "code",
    "id": "rX-brDlaeFNZ",
    "outputId": "9f7ac2f2-a298-4ae2-9977-6cc5c100f7c3"
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "The losses are [1000.0, 1000.0, 1000.0, 1000.0].\n",
      "\n",
      "The loss is 4000.0. This took 0.472 seconds. Run the next cell to see if the exercise was done correctly.\n"
     ]
    }
   ],
   "source": [
    "start_time = time.time()\n",
    "losses = []\n",
    "for filename in ['file1', 'file2', 'file3', 'file4']:\n",
    "    inner_start = time.time()\n",
    "\n",
    "    data = load_data.remote(filename)\n",
    "    normalized_data = normalize_data.remote(data)\n",
    "    features = extract_features.remote(normalized_data)\n",
    "    loss = compute_loss.remote(features)\n",
    "    losses.append(loss)\n",
    "    \n",
    "    inner_end = time.time()\n",
    "    \n",
    "    if inner_end - inner_start >= 0.1:\n",
    "        raise Exception('You may be calling ray.get inside of the for loop! '\n",
    "                        'Doing this will prevent parallelism from being exposed. '\n",
    "                        'Make sure to only call ray.get once outside of the for loop.')\n",
    "\n",
    "losses = ray.get(losses)\n",
    "\n",
    "print('The losses are {}.'.format(losses) + '\\n')\n",
    "loss = sum(losses)\n",
    "\n",
    "duration = time.time() - start_time\n",
    "\n",
    "print('The loss is {}. This took {:.3f} seconds. Run the next cell to see '\n",
    "      'if the exercise was done correctly.'.format(loss, duration))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "colab_type": "text",
    "id": "nOevdR4LeLlk"
   },
   "source": [
    "**VERIFY:** Run some checks to verify that the changes you made to the code were correct. Some of the checks should fail when you initially run the cells. After completing the exercises, the checks should pass."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/",
     "height": 34
    },
    "colab_type": "code",
    "id": "FlTAyw09eJSL",
    "outputId": "e3c884ca-c500-4b84-acb5-4b5e73224d9f"
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Success! The example took 0.472 seconds.\n"
     ]
    }
   ],
   "source": [
    "assert loss == 4000\n",
    "assert duration < 0.8, ('The loop took {:.3f} seconds. This is too slow.'\n",
    "                        .format(duration))\n",
    "assert duration > 0.4, ('The loop took {:.3f} seconds. This is too fast.'\n",
    "                        .format(duration))\n",
    "\n",
    "print('Success! The example took {:.3f} seconds.'.format(duration))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "colab_type": "text",
    "id": "r1fHJZWDeRGb"
   },
   "source": [
    "**EXERCISE:** Use the UI to view the task timeline and to verify that the four tasks were executed in parallel. You can do this as follows.\n",
    "\n",
    "1. Run the following cell to generate a JSON file containing the profiling data.\n",
    "2. Download the timeline file by right clicking on `exercise_2.json` in the **Files** tab in the navigator to the left, right clicking, and selecting  **\"Download\"**.\n",
    "3. Enter **chrome://tracing** into the Chrome web browser, click on the **\"Load\"** button, and select the downloaded JSON file.\n",
    "\n",
    "To navigate within the timeline:\n",
    "- Move around by clicking and dragging.\n",
    "- Zoom in and out by holding **alt** on Windows or **option** on Mac and scrolling.\n",
    "\n",
    "**NOTE:** The timeline visualization will only work in **Chrome**."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "colab": {},
    "colab_type": "code",
    "id": "R96y_U99eRdg"
   },
   "outputs": [],
   "source": [
    "ray.timeline(filename=\"exercise_2.json\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "colab_type": "text",
    "id": "4UMNWDWOeVI2"
   },
   "source": [
    "### Application: Parallel web-scraping\n",
    "\n",
    "One useful application of what we have learned so far is to scrape information from the web. We will illustrate this in a toy setting, but the same principles apply on a large scale where crawling through websites, parsing them and extracting useful content (e.g. for building a search index or populating a database) is often very computationally demanding.\n",
    "\n",
    "We break up the process into multiple steps. We first grab the raw HTML of the website using Python's requests package. Then, we use BeautifulSoup to parse the HTML to find the relevant information. Finally, we populate a pandas DataFrames so that we are able to work with the data.\n",
    "\n",
    "To demonstrate this, we scrape GitHub commits to see the latest commits on several repositories."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "colab": {},
    "colab_type": "code",
    "id": "6apTGnCZeTDA"
   },
   "outputs": [],
   "source": [
    "from bs4 import BeautifulSoup\n",
    "import requests\n",
    "\n",
    "import pandas as pd"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "colab_type": "text",
    "id": "IYW3fS-Aeg2s"
   },
   "source": [
    "The following function uses these libraries to parse the latest commits from several repositories on GitHub."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "colab": {},
    "colab_type": "code",
    "id": "GXRdG9zYehhI"
   },
   "outputs": [],
   "source": [
    "@ray.remote\n",
    "def fetch_commits(repo):\n",
    "    url = 'https://github.com/{}/commits/master'.format(repo)\n",
    "    response = requests.get(url)\n",
    "    soup = BeautifulSoup(response.text, 'lxml')\n",
    "    df = pd.DataFrame(columns=['title', 'link'])\n",
    "    for g in soup.find_all(class_='commit-title'):\n",
    "        entry = {}\n",
    "        title = g.find_all(class_='message')[0]['aria-label']\n",
    "        entry['title'] = title\n",
    "        links = g.find_all(class_='issue-link')\n",
    "        if len(links) >= 1:\n",
    "            link = links[0]['data-url']\n",
    "            entry['link'] = link\n",
    "        df = df.append(pd.DataFrame(entry, index=[0]), sort=False)\n",
    "    \n",
    "    df['repository'] = repo\n",
    "    return df"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "colab_type": "text",
    "id": "zjhbi8UXen7d"
   },
   "source": [
    "Let's try this out to get results for some ray related topics serially."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/",
     "height": 34
    },
    "colab_type": "code",
    "id": "XENYO57kepSU",
    "outputId": "8ef708a8-ce71-4f5d-af26-9037ff458c63"
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Constructing the dataframe took 1.582 seconds.\n"
     ]
    }
   ],
   "source": [
    "start = time.time()\n",
    "repos = [\"ray-project/ray\", \"modin-project/modin\", \"tensorflow/tensorflow\", \"apache/arrow\"]\n",
    "results = []\n",
    "for repo in repos:\n",
    "    df = fetch_commits.remote(repo)\n",
    "    results.append(df)\n",
    "    \n",
    "df = pd.concat(ray.get(results), sort=False)\n",
    "duration = time.time() - start\n",
    "print(\"Constructing the dataframe took {:.3f} seconds.\".format(duration))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "colab_type": "text",
    "id": "eZ5VmqgoerPb"
   },
   "source": [
    "**EXERCISE**: Speed up the above serial query by making `fetch_commits` a remote function in order to scrape GitHub results in parallel. Then, see a sample of the data scraped below and feel free to play with the data to find other resources to learn more about these libraries!"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/",
     "height": 1000
    },
    "colab_type": "code",
    "id": "58RIGdF1etXY",
    "outputId": "08367e0e-90e8-4e98-8631-ec04db5b87d9"
   },
   "outputs": [
    {
     "data": {
      "text/html": [
       "<div>\n",
       "<style scoped>\n",
       "    .dataframe tbody tr th:only-of-type {\n",
       "        vertical-align: middle;\n",
       "    }\n",
       "\n",
       "    .dataframe tbody tr th {\n",
       "        vertical-align: top;\n",
       "    }\n",
       "\n",
       "    .dataframe thead th {\n",
       "        text-align: right;\n",
       "    }\n",
       "</style>\n",
       "<table border=\"1\" class=\"dataframe\">\n",
       "  <thead>\n",
       "    <tr style=\"text-align: right;\">\n",
       "      <th></th>\n",
       "      <th>title</th>\n",
       "      <th>link</th>\n",
       "      <th>repository</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>[Serve] Remove handle passing in tail recursio...</td>\n",
       "      <td>https://github.com/ray-project/ray/issues/5894</td>\n",
       "      <td>ray-project/ray</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>rllib: use pytorch's fn to see if gpu is avail...</td>\n",
       "      <td>https://github.com/ray-project/ray/issues/5890</td>\n",
       "      <td>ray-project/ray</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>[minor][docs] Remove example link (#5880)</td>\n",
       "      <td>https://github.com/ray-project/ray/issues/5880</td>\n",
       "      <td>ray-project/ray</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>Don't wrap RayError with RayTaskError (#5870)</td>\n",
       "      <td>https://github.com/ray-project/ray/issues/5870</td>\n",
       "      <td>ray-project/ray</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>[autoscaler] Fix quoting (#5891)</td>\n",
       "      <td>https://github.com/ray-project/ray/issues/5891</td>\n",
       "      <td>ray-project/ray</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>[Serve] Hotfix: Fix actor handle hashing in me...</td>\n",
       "      <td>https://github.com/ray-project/ray/issues/5886</td>\n",
       "      <td>ray-project/ray</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>Python 2 compatibility. (#5887)</td>\n",
       "      <td>https://github.com/ray-project/ray/issues/5887</td>\n",
       "      <td>ray-project/ray</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>Fix str of RayTaskError (#5878)\\n\\n* fix key e...</td>\n",
       "      <td>https://github.com/ray-project/ray/issues/5878</td>\n",
       "      <td>ray-project/ray</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>Fix linux wheel build (#5881)</td>\n",
       "      <td>https://github.com/ray-project/ray/issues/5881</td>\n",
       "      <td>ray-project/ray</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>[Dashboard] Improve handling of logs and error...</td>\n",
       "      <td>https://github.com/ray-project/ray/issues/5857</td>\n",
       "      <td>ray-project/ray</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>Implement fair task queueing to prevent task s...</td>\n",
       "      <td>https://github.com/ray-project/ray/issues/5851</td>\n",
       "      <td>ray-project/ray</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>[tune][minor] formatting examples, fix travis ...</td>\n",
       "      <td>https://github.com/ray-project/ray/issues/5869</td>\n",
       "      <td>ray-project/ray</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>[tune] Readable trial progress output (#5822)\\...</td>\n",
       "      <td>https://github.com/ray-project/ray/issues/5822</td>\n",
       "      <td>ray-project/ray</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>temporarily remove tensorflow test (#5866)</td>\n",
       "      <td>https://github.com/ray-project/ray/issues/5866</td>\n",
       "      <td>ray-project/ray</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>[doc] Hyperparameter Tuning Gallery Entry (#57...</td>\n",
       "      <td>https://github.com/ray-project/ray/issues/5786</td>\n",
       "      <td>ray-project/ray</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>Qmix on gpu and with non-stacked-obs environme...</td>\n",
       "      <td>https://github.com/ray-project/ray/issues/5751</td>\n",
       "      <td>ray-project/ray</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>Fix actor ID collision in local mode (#5863)\\n...</td>\n",
       "      <td>https://github.com/ray-project/ray/issues/5863</td>\n",
       "      <td>ray-project/ray</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>[tune] Check node liveness before result fetch...</td>\n",
       "      <td>https://github.com/ray-project/ray/issues/5844</td>\n",
       "      <td>ray-project/ray</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>[tune] MedianStopping on result (#5402)\\n\\n* a...</td>\n",
       "      <td>https://github.com/ray-project/ray/issues/5402</td>\n",
       "      <td>ray-project/ray</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>Link to kubernetes config files in docs (#5865)</td>\n",
       "      <td>https://github.com/ray-project/ray/issues/5865</td>\n",
       "      <td>ray-project/ray</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>Fix class attributes and methods for actor cla...</td>\n",
       "      <td>https://github.com/ray-project/ray/issues/5802</td>\n",
       "      <td>ray-project/ray</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>[core worker] Submit Python actor tasks throug...</td>\n",
       "      <td>https://github.com/ray-project/ray/issues/5750</td>\n",
       "      <td>ray-project/ray</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>Fix TF2 / rllib test (#5846)</td>\n",
       "      <td>https://github.com/ray-project/ray/issues/5846</td>\n",
       "      <td>ray-project/ray</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>[Serve] Implement metric interface (#5852)\\n\\n...</td>\n",
       "      <td>https://github.com/ray-project/ray/issues/5852</td>\n",
       "      <td>ray-project/ray</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>[Serve] Implement replica scaling (#5850)\\n\\n*...</td>\n",
       "      <td>https://github.com/ray-project/ray/issues/5850</td>\n",
       "      <td>ray-project/ray</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>Restore support for Python 3.5 (#5818)\\n\\n* Ad...</td>\n",
       "      <td>https://github.com/ray-project/ray/issues/5818</td>\n",
       "      <td>ray-project/ray</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>[Serve] Implement flask_request and named pyth...</td>\n",
       "      <td>https://github.com/ray-project/ray/issues/5849</td>\n",
       "      <td>ray-project/ray</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>[tune] PBT + Memnn example (#5723)\\n\\n* Add ex...</td>\n",
       "      <td>https://github.com/ray-project/ray/issues/5723</td>\n",
       "      <td>ray-project/ray</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>Fix obs space lo/hi (#5826)</td>\n",
       "      <td>https://github.com/ray-project/ray/issues/5826</td>\n",
       "      <td>ray-project/ray</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>Just die on signal (#5842)</td>\n",
       "      <td>https://github.com/ray-project/ray/issues/5842</td>\n",
       "      <td>ray-project/ray</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>...</th>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>ARROW-6661: [Java] Implement APIs like slice t...</td>\n",
       "      <td>NaN</td>\n",
       "      <td>apache/arrow</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>ARROW-6074: [FlightRPC][Java] Middleware\\n\\n(T...</td>\n",
       "      <td>NaN</td>\n",
       "      <td>apache/arrow</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>ARROW-6732: [Java] Implement quick sort in a n...</td>\n",
       "      <td>NaN</td>\n",
       "      <td>apache/arrow</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>ARROW-6721: [JAVA] Avro adapter benchmark only...</td>\n",
       "      <td>NaN</td>\n",
       "      <td>apache/arrow</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>ARROW-1638: [Java] IPC roundtrip for null type...</td>\n",
       "      <td>NaN</td>\n",
       "      <td>apache/arrow</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>ARROW-6711: [C++] Consolidate Filter and Expre...</td>\n",
       "      <td>NaN</td>\n",
       "      <td>apache/arrow</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>ARROW-6835: [Archery][CMake] Restore ARROW_LIN...</td>\n",
       "      <td>NaN</td>\n",
       "      <td>apache/arrow</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>ARROW-6855: [FlightRPC][C++][Python] Flight mi...</td>\n",
       "      <td>NaN</td>\n",
       "      <td>apache/arrow</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>ARROW-6833: [R][CI] Add crossbow job for full ...</td>\n",
       "      <td>NaN</td>\n",
       "      <td>apache/arrow</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>ARROW-6831: [R] Update R macOS/Windows builds ...</td>\n",
       "      <td>NaN</td>\n",
       "      <td>apache/arrow</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>ARROW-6832: [R] Implement Codec::IsAvailable\\n...</td>\n",
       "      <td>NaN</td>\n",
       "      <td>apache/arrow</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>ARROW-6807: [Java][FlightRPC] Expose gRPC serv...</td>\n",
       "      <td>NaN</td>\n",
       "      <td>apache/arrow</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>ARROW-6782: [C++] Do not require Boost for min...</td>\n",
       "      <td>NaN</td>\n",
       "      <td>apache/arrow</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>ARROW-6834: [C++][TRIAGE] Pin gtest version 1....</td>\n",
       "      <td>NaN</td>\n",
       "      <td>apache/arrow</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>[Python][Minor] Fix typo in function docstring</td>\n",
       "      <td>NaN</td>\n",
       "      <td>apache/arrow</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>ARROW-6274: [Rust] [DataFusion] Add support fo...</td>\n",
       "      <td>NaN</td>\n",
       "      <td>apache/arrow</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>ARROW-6631: [C++] Do not build any compression...</td>\n",
       "      <td>NaN</td>\n",
       "      <td>apache/arrow</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>ARROW-6466: [Integration][CI] Move integration...</td>\n",
       "      <td>NaN</td>\n",
       "      <td>apache/arrow</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>ARROW-6778: [C++] Support cast for DurationTyp...</td>\n",
       "      <td>NaN</td>\n",
       "      <td>apache/arrow</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>ARROW-6764: [C++] Create a readahead iterator\\...</td>\n",
       "      <td>NaN</td>\n",
       "      <td>apache/arrow</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>ARROW-6824: [Plasma] Allow creation of multipl...</td>\n",
       "      <td>NaN</td>\n",
       "      <td>apache/arrow</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>ARROW-6812: [Java] Fix License header\\n\\nRemov...</td>\n",
       "      <td>NaN</td>\n",
       "      <td>apache/arrow</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>ARROW-6321: [Python] Ability to create Extensi...</td>\n",
       "      <td>NaN</td>\n",
       "      <td>apache/arrow</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>ARROW-6803: [Rust] [DataFusion] Performance op...</td>\n",
       "      <td>NaN</td>\n",
       "      <td>apache/arrow</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>ARROW-5802: [CI][Archery] Dockerify lint utili...</td>\n",
       "      <td>NaN</td>\n",
       "      <td>apache/arrow</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>ARROW-6768: [C++][Dataset] Add method to conve...</td>\n",
       "      <td>NaN</td>\n",
       "      <td>apache/arrow</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>ARROW-6695: [Rust] [DataFusion] Remove legacy ...</td>\n",
       "      <td>NaN</td>\n",
       "      <td>apache/arrow</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>ARROW-5655: [Python] Table.from_pydict/from_ar...</td>\n",
       "      <td>NaN</td>\n",
       "      <td>apache/arrow</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>ARROW-6811: [R] Assorted post-0.15 release cle...</td>\n",
       "      <td>NaN</td>\n",
       "      <td>apache/arrow</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>[Ruby][Docs] Update link to Apache Parquet C++...</td>\n",
       "      <td>https://github.com/apache/arrow/issues/5604</td>\n",
       "      <td>apache/arrow</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "<p>140 rows × 3 columns</p>\n",
       "</div>"
      ],
      "text/plain": [
       "                                                title  ...       repository\n",
       "0   [Serve] Remove handle passing in tail recursio...  ...  ray-project/ray\n",
       "0   rllib: use pytorch's fn to see if gpu is avail...  ...  ray-project/ray\n",
       "0           [minor][docs] Remove example link (#5880)  ...  ray-project/ray\n",
       "0       Don't wrap RayError with RayTaskError (#5870)  ...  ray-project/ray\n",
       "0                    [autoscaler] Fix quoting (#5891)  ...  ray-project/ray\n",
       "0   [Serve] Hotfix: Fix actor handle hashing in me...  ...  ray-project/ray\n",
       "0                     Python 2 compatibility. (#5887)  ...  ray-project/ray\n",
       "0   Fix str of RayTaskError (#5878)\\n\\n* fix key e...  ...  ray-project/ray\n",
       "0                       Fix linux wheel build (#5881)  ...  ray-project/ray\n",
       "0   [Dashboard] Improve handling of logs and error...  ...  ray-project/ray\n",
       "0   Implement fair task queueing to prevent task s...  ...  ray-project/ray\n",
       "0   [tune][minor] formatting examples, fix travis ...  ...  ray-project/ray\n",
       "0   [tune] Readable trial progress output (#5822)\\...  ...  ray-project/ray\n",
       "0          temporarily remove tensorflow test (#5866)  ...  ray-project/ray\n",
       "0   [doc] Hyperparameter Tuning Gallery Entry (#57...  ...  ray-project/ray\n",
       "0   Qmix on gpu and with non-stacked-obs environme...  ...  ray-project/ray\n",
       "0   Fix actor ID collision in local mode (#5863)\\n...  ...  ray-project/ray\n",
       "0   [tune] Check node liveness before result fetch...  ...  ray-project/ray\n",
       "0   [tune] MedianStopping on result (#5402)\\n\\n* a...  ...  ray-project/ray\n",
       "0     Link to kubernetes config files in docs (#5865)  ...  ray-project/ray\n",
       "0   Fix class attributes and methods for actor cla...  ...  ray-project/ray\n",
       "0   [core worker] Submit Python actor tasks throug...  ...  ray-project/ray\n",
       "0                        Fix TF2 / rllib test (#5846)  ...  ray-project/ray\n",
       "0   [Serve] Implement metric interface (#5852)\\n\\n...  ...  ray-project/ray\n",
       "0   [Serve] Implement replica scaling (#5850)\\n\\n*...  ...  ray-project/ray\n",
       "0   Restore support for Python 3.5 (#5818)\\n\\n* Ad...  ...  ray-project/ray\n",
       "0   [Serve] Implement flask_request and named pyth...  ...  ray-project/ray\n",
       "0   [tune] PBT + Memnn example (#5723)\\n\\n* Add ex...  ...  ray-project/ray\n",
       "0                         Fix obs space lo/hi (#5826)  ...  ray-project/ray\n",
       "0                          Just die on signal (#5842)  ...  ray-project/ray\n",
       "..                                                ...  ...              ...\n",
       "0   ARROW-6661: [Java] Implement APIs like slice t...  ...     apache/arrow\n",
       "0   ARROW-6074: [FlightRPC][Java] Middleware\\n\\n(T...  ...     apache/arrow\n",
       "0   ARROW-6732: [Java] Implement quick sort in a n...  ...     apache/arrow\n",
       "0   ARROW-6721: [JAVA] Avro adapter benchmark only...  ...     apache/arrow\n",
       "0   ARROW-1638: [Java] IPC roundtrip for null type...  ...     apache/arrow\n",
       "0   ARROW-6711: [C++] Consolidate Filter and Expre...  ...     apache/arrow\n",
       "0   ARROW-6835: [Archery][CMake] Restore ARROW_LIN...  ...     apache/arrow\n",
       "0   ARROW-6855: [FlightRPC][C++][Python] Flight mi...  ...     apache/arrow\n",
       "0   ARROW-6833: [R][CI] Add crossbow job for full ...  ...     apache/arrow\n",
       "0   ARROW-6831: [R] Update R macOS/Windows builds ...  ...     apache/arrow\n",
       "0   ARROW-6832: [R] Implement Codec::IsAvailable\\n...  ...     apache/arrow\n",
       "0   ARROW-6807: [Java][FlightRPC] Expose gRPC serv...  ...     apache/arrow\n",
       "0   ARROW-6782: [C++] Do not require Boost for min...  ...     apache/arrow\n",
       "0   ARROW-6834: [C++][TRIAGE] Pin gtest version 1....  ...     apache/arrow\n",
       "0      [Python][Minor] Fix typo in function docstring  ...     apache/arrow\n",
       "0   ARROW-6274: [Rust] [DataFusion] Add support fo...  ...     apache/arrow\n",
       "0   ARROW-6631: [C++] Do not build any compression...  ...     apache/arrow\n",
       "0   ARROW-6466: [Integration][CI] Move integration...  ...     apache/arrow\n",
       "0   ARROW-6778: [C++] Support cast for DurationTyp...  ...     apache/arrow\n",
       "0   ARROW-6764: [C++] Create a readahead iterator\\...  ...     apache/arrow\n",
       "0   ARROW-6824: [Plasma] Allow creation of multipl...  ...     apache/arrow\n",
       "0   ARROW-6812: [Java] Fix License header\\n\\nRemov...  ...     apache/arrow\n",
       "0   ARROW-6321: [Python] Ability to create Extensi...  ...     apache/arrow\n",
       "0   ARROW-6803: [Rust] [DataFusion] Performance op...  ...     apache/arrow\n",
       "0   ARROW-5802: [CI][Archery] Dockerify lint utili...  ...     apache/arrow\n",
       "0   ARROW-6768: [C++][Dataset] Add method to conve...  ...     apache/arrow\n",
       "0   ARROW-6695: [Rust] [DataFusion] Remove legacy ...  ...     apache/arrow\n",
       "0   ARROW-5655: [Python] Table.from_pydict/from_ar...  ...     apache/arrow\n",
       "0   ARROW-6811: [R] Assorted post-0.15 release cle...  ...     apache/arrow\n",
       "0   [Ruby][Docs] Update link to Apache Parquet C++...  ...     apache/arrow\n",
       "\n",
       "[140 rows x 3 columns]"
      ]
     },
     "execution_count": 22,
     "metadata": {
      "tags": []
     },
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "colab_type": "text",
    "id": "z-c-u7Nvewci"
   },
   "source": [
    "# Exercise 3 - Nested Parallelism\n",
    "\n",
    "**GOAL:** The goal of this exercise is to show how to create nested tasks by calling a remote function inside of another remote function.\n",
    "\n",
    "In this exercise, you will implement the structure of a parallel hyperparameter sweep which trains a number of models in parallel. Each model will be trained using parallel gradient computations.\n",
    "\n",
    "### Concepts for this Exercise - Nested Remote Functions\n",
    "\n",
    "Remote functions can call other functions. For example, consider the following.\n",
    "\n",
    "```python\n",
    "@ray.remote\n",
    "def f():\n",
    "    return 1\n",
    "\n",
    "@ray.remote\n",
    "def g():\n",
    "    # Call f 4 times and return the resulting object IDs.\n",
    "    results = []\n",
    "    for _ in range(4):\n",
    "      results.append(f.remote())\n",
    "    return results\n",
    "\n",
    "@ray.remote\n",
    "def h():\n",
    "    # Call f 4 times, block until those 4 tasks finish,\n",
    "    # retrieve the results, and return the values.\n",
    "    results = []\n",
    "    for _ in range(4):\n",
    "      results.append(f.remote())\n",
    "    return ray.get(results)\n",
    "```\n",
    "\n",
    "Then calling `g` and `h` produces the following behavior.\n",
    "\n",
    "```python\n",
    ">>> ray.get(g.remote())\n",
    "[ObjectID(b1457ba0911ae84989aae86f89409e953dd9a80e),\n",
    " ObjectID(7c14a1d13a56d8dc01e800761a66f09201104275),\n",
    " ObjectID(99763728ffc1a2c0766a2000ebabded52514e9a6),\n",
    " ObjectID(9c2f372e1933b04b2936bb6f58161285829b9914)]\n",
    "\n",
    ">>> ray.get(h.remote())\n",
    "[1, 1, 1, 1]\n",
    "```"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "colab": {},
    "colab_type": "code",
    "id": "ppUDwvBMe0ue"
   },
   "outputs": [],
   "source": [
    "from __future__ import absolute_import\n",
    "from __future__ import division\n",
    "from __future__ import print_function\n",
    "\n",
    "import numpy as np\n",
    "import time\n",
    "\n",
    "import ray"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 30,
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/",
     "height": 34
    },
    "colab_type": "code",
    "id": "2aEuUdJDe-uY",
    "outputId": "ec958263-c3ad-44a0-94eb-baa9d7490f81"
   },
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "2019-10-13 10:07:32,295\tERROR worker.py:1432 -- Calling ray.init() again after it has already been called.\n"
     ]
    }
   ],
   "source": [
    "ray.init(num_cpus=9, ignore_reinit_error=True)\n",
    "\n",
    "# Sleep a little to improve the accuracy of the timing measurements used below,\n",
    "# because some workers may still be starting up in the background.\n",
    "time.sleep(2.0)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "colab_type": "text",
    "id": "RKX1DnmbfAJv"
   },
   "source": [
    "This example represents a hyperparameter sweep in which multiple models are trained in parallel. Each model training task also performs data parallel gradient computations.\n",
    "\n",
    "**EXERCISE:** Turn `compute_gradient` and `train_model` into remote functions so that they can be executed in parallel. Inside of `train_model`, do the calls to `compute_gradient` in parallel and fetch the results using `ray.get`."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "colab": {},
    "colab_type": "code",
    "id": "GAdMoSp7fBxV"
   },
   "outputs": [],
   "source": [
    "@ray.remote\n",
    "def compute_gradient(data, current_model):\n",
    "    time.sleep(0.03)\n",
    "    return 1\n",
    "\n",
    "@ray.remote\n",
    "def train_model(hyperparameters):\n",
    "    current_model = 0\n",
    "    # Iteratively improve the current model. This outer loop cannot be parallelized.\n",
    "    for _ in range(10):\n",
    "        # EXERCISE: Parallelize the list comprehension in the line below. After you\n",
    "        # turn \"compute_gradient\" into a remote function, you will need to call it\n",
    "        # with \".remote\". The results must be retrieved with \"ray.get\" before \"sum\"\n",
    "        # is called.\n",
    "        total_gradient = sum(ray.get([compute_gradient.remote(j, current_model) for j in range(2)]))\n",
    "        current_model += total_gradient\n",
    "\n",
    "    return current_model\n",
    "\n",
    "assert hasattr(compute_gradient, 'remote'), 'compute_gradient must be a remote function'\n",
    "assert hasattr(train_model, 'remote'), 'train_model must be a remote function'"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "colab_type": "text",
    "id": "6yR2xgWhfEK3"
   },
   "source": [
    "**EXERCISE:** The code below runs 3 hyperparameter experiments. Change this to run the experiments in parallel."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "colab": {},
    "colab_type": "code",
    "id": "KQhgStdSfGMt"
   },
   "outputs": [],
   "source": [
    "# Sleep a little to improve the accuracy of the timing measurements below.\n",
    "time.sleep(2.0)\n",
    "start_time = time.time()\n",
    "\n",
    "# Run some hyperparaameter experiments.\n",
    "results = []\n",
    "for hyperparameters in [{'learning_rate': 1e-1, 'batch_size': 100},\n",
    "                        {'learning_rate': 1e-2, 'batch_size': 100},\n",
    "                        {'learning_rate': 1e-3, 'batch_size': 100}]:\n",
    "    results.append(train_model.remote(hyperparameters))\n",
    "\n",
    "# EXERCISE: Once you've turned \"results\" into a list of Ray ObjectIDs\n",
    "# by calling train_model.remote, you will need to turn \"results\" back\n",
    "# into a list of integers, e.g., by doing \"results = ray.get(results)\".\n",
    "\n",
    "results = ray.get(results)\n",
    "\n",
    "end_time = time.time()\n",
    "duration = end_time - start_time\n",
    "\n",
    "assert all([isinstance(x, int) for x in results]), \\\n",
    "    'Looks like \"results\" is {}. You may have forgotten to call ray.get.'.format(results)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "colab_type": "text",
    "id": "MZbRvJ9JfJcl"
   },
   "source": [
    "**VERIFY:** Run some checks to verify that the changes you made to the code were correct. Some of the checks should fail when you initially run the cells. After completing the exercises, the checks should pass.\n",
    "\n",
    "**NOTE:** This exercise is known to have issues running on remote notebooks that can be resolved by rerunning the above cell a second time."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 36,
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/",
     "height": 34
    },
    "colab_type": "code",
    "id": "BN6QZZGEfLAu",
    "outputId": "6ebec9bb-7d13-402c-f283-bc0f471a599c"
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Success! The example took 0.497 seconds.\n"
     ]
    }
   ],
   "source": [
    "assert results == [20, 20, 20]\n",
    "assert duration < 0.5, ('The experiments ran in {:.3f} seconds. This is too '\n",
    "                         'slow.'.format(duration))\n",
    "assert duration > 0.3, ('The experiments ran in {:.3f} seconds. This is too '\n",
    "                        'fast.'.format(duration))\n",
    "\n",
    "print('Success! The example took {:.3f} seconds.'.format(duration))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "colab_type": "text",
    "id": "Xuf2d5mrfMms"
   },
   "source": [
    "**EXERCISE:** Use the UI to view the task timeline and to verify that the four tasks were executed in parallel. You can do this as follows.\n",
    "\n",
    "1. Run the following cell to generate a JSON file containing the profiling data.\n",
    "2. Download the timeline file by right clicking on `exercise_3.json` in the **Files** tab in the navigator to the left, right clicking, and selecting  **\"Download\"**.\n",
    "3. Enter **chrome://tracing** into the Chrome web browser, click on the **\"Load\"** button, and select the downloaded JSON file.\n",
    "\n",
    "To navigate within the timeline:\n",
    "- Move around by clicking and dragging.\n",
    "- Zoom in and out by holding **alt** on Windows or **option** on Mac and scrolling.\n",
    "\n",
    "**NOTE:** The timeline visualization will only work in **Chrome**."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "colab": {},
    "colab_type": "code",
    "id": "7WoR39fnfOlx"
   },
   "outputs": [],
   "source": [
    "ray.timeline(filename=\"exercise_3.json\")"
   ]
  }
 ],
 "metadata": {
  "colab": {
   "name": "[ANSWER KEY] Ray Tutorial - Remote Functions",
   "provenance": [],
   "toc_visible": true
  },
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.7.4"
  },
  "toc": {
   "base_numbering": 1,
   "nav_menu": {},
   "number_sections": false,
   "sideBar": true,
   "skip_h1_title": false,
   "title_cell": "Table of Contents",
   "title_sidebar": "Contents",
   "toc_cell": false,
   "toc_position": {
    "height": "calc(100% - 180px)",
    "left": "10px",
    "top": "150px",
    "width": "382.391px"
   },
   "toc_section_display": true,
   "toc_window_display": true
  }
 },
 "nbformat": 4,
 "nbformat_minor": 1
}
